#![allow(unused_imports)]
#![allow(unused_variables)]
#![allow(dead_code)]

use std::collections::HashMap ;
use std::iter ;
use nonempty::NonEmpty;
use anyhow::{Result, anyhow} ;
use serde::{Deserialize, Serialize, Deserializer, Serializer, de::Visitor} ;
use serde_json::Value as JsonValue ;
use itertools::Itertools ;
use tracing::{info, debug, trace} ;

#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
pub enum DefaultValue {
    DNull,
    DBoolean(bool),
    DInt(Schema, i32),
    DLong(Schema, i64),
    DFloat(Schema, f32),
    DDouble(Schema, f64),
    DBytes(Schema, Vec<u8>),
    DString(Schema, String),
    DArray(Vec<DefaultValue>),
    DMap(HashMap<String, DefaultValue>),
    DRecord(Schema, HashMap<String, DefaultValue>),
    DUnion(Vec<Schema>, Schema, Box<DefaultValue>),
    DFixed(Schema, Vec<u8>),
    DEnum(Schema, usize, String)
}

#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
pub enum Schema {
    Null,
    Boolean,
    Int { logical_type_i : Option<LogicalTypeInt> },
    Long { logical_type_l : Option<LogicalTypeLong> },
    Float,
    Double,
    Bytes { logical_type_b : Option<LogicalTypeBytes> },
    String { logical_type_s : Option<LogicalTypeString> },
    Array { item: Box<Schema> },
    Map { values: Box<Schema> },
    NamedType(TypeName),
    Record {
        name: TypeName,
        aliases: Vec<TypeName>,
        doc: Option<String>,
        fields: Vec<Field>
    },
    Enum {
        name: TypeName,
        aliases: Vec<TypeName>,
        doc: Option<String>,
        symbols: Vec<String>,
    },
    Union {
        options: Vec<Schema>,
    },
    Fixed {
        name: TypeName,
        aliases: Vec<TypeName>,
        size: u64,
        logical_type_f: Option<LogicalTypeFixed>,
    }
}

#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
pub struct Field {
    fld_name: String,
    fld_aliases: Vec<String>,
    fld_doc: Option<String>,
    fld_order: Option<Order>,
    fld_type: Schema,
    fld_default: Option<DefaultValue>,
}

#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum Order {
    Ascending,
    Descending,
    Ignore,
}
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
pub enum LogicalTypeBytes {
    DecimalB(Decimal),
}
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
pub enum LogicalTypeFixed {
    DecimalF(Decimal),
    Duration,
}
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
pub enum LogicalTypeInt {
    DecimalI(Decimal),
    Date,
    TimeMillis,
}
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
pub enum LogicalTypeLong {
    DecimalL(Decimal),
    TimeMicros,
    TimestampMillis,
    TimestampMicros,
    LocalTimestampMillis,
    LocalTimestampMicros,
}
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
pub enum LogicalTypeString {
    UUID,
}

#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
pub struct Decimal {
    precision: u64,
    scale: u64
}

// impl Eq for Schema {}

pub fn mk_enum(name: TypeName, aliases: Vec<TypeName>, doc: Option<String>, symbols: Vec<String>) -> Schema {
    Schema::Enum { name, aliases, doc, symbols }
}

pub fn mk_union(options: NonEmpty<Schema>) -> Schema {
    Schema::Union {  options: options.into() }
}

#[derive(Debug, Clone, PartialEq, Eq, Hash, Serialize, Deserialize)]
pub struct TypeName {
    base_name: String,
    namespace: Vec<String>,
}


pub fn render_fullname(type_name: &TypeName) -> String {
    let TypeName {base_name, namespace, ..} = type_name ;
    namespace.iter().chain(std::iter::once(base_name)).join(".")
}

pub fn parse_fullname(s: &str) -> TypeName {
    let mut xs = s.split('.').into_iter() ;
    let base_name = xs.next().unwrap().to_owned() ;
    let namespace = xs.map(|it| it.to_owned()).filter(|x| x != "").collect_vec() ;
    TypeName { base_name, namespace } 
}

pub fn mk_type_name(context: Option<&TypeName>, name: &str, ns: Option<&str>) -> TypeName {
    fn is_fullname(name: &str) -> bool {
        name.contains(".")
    }
    if is_fullname(name) {
        parse_fullname(name)
    } else {
        match ns {
            Some(ns) => TypeName{
                base_name: name.to_string(), 
                namespace: ns.split(".").filter(|it| it != &"").map(|it| it.to_string()).collect_vec()
            },
            None => TypeName{ base_name: name.to_string(), namespace: context.map_or(vec![], |v| v.namespace.clone()) }
        }
    }
}

pub fn type_name(bt: &Schema) -> String { 
    fn decimal_name(decimal: &Decimal) -> String {
        let Decimal{precision, scale} = decimal ;
        format!("decimal({},{})", precision, scale)
    }

    match bt {
        Schema::Null => "null".to_string(),
        Schema::Boolean => "boolean".to_string(),
        Schema::Int{logical_type_i: None} => "int".to_string(),
        Schema::Int{logical_type_i: Some(LogicalTypeInt::DecimalI(d))} => decimal_name(d),
        Schema::Int{logical_type_i: Some(LogicalTypeInt::Date)} => "date".to_string(),
        Schema::Int{logical_type_i: Some(LogicalTypeInt::TimeMillis)} => "time-millis".to_string(),
        Schema::Long{logical_type_l: None} => "long".to_string(),
        Schema::Long{logical_type_l: Some(LogicalTypeLong::DecimalL(d))} => decimal_name(d),
        Schema::Long{logical_type_l: Some(LogicalTypeLong::TimeMicros)} => "time-micros".to_string(),
        Schema::Long{logical_type_l: Some(LogicalTypeLong::TimestampMillis)} => "timestamp-millis".to_string(),
        Schema::Long{logical_type_l: Some(LogicalTypeLong::TimestampMicros)} => "timestamp-micros".to_string(),
        Schema::Long{logical_type_l: Some(LogicalTypeLong::LocalTimestampMillis)} => "local-timestamp-millis".to_string(),
        Schema::Long{logical_type_l: Some(LogicalTypeLong::LocalTimestampMicros)} => "local-timestamp-macros".to_string(),
        Schema::Float => "float".to_string(),
        Schema::Double => "double".to_string(),
        Schema::Bytes{logical_type_b: None} => "bytes".to_string(),
        Schema::Bytes{logical_type_b: Some(LogicalTypeBytes::DecimalB(d))} => decimal_name(d),
        Schema::String{logical_type_s: None} => "string".to_string(),
        Schema::String{logical_type_s: Some(LogicalTypeString::UUID)} => "uuid".to_string(),
        Schema::Array{..} => "array".to_string(),
        Schema::Map{..} => "map".to_string(),
        Schema::Record{name, ..} => render_fullname(name),
        Schema::NamedType(name) => render_fullname(name),
        Schema::Union{options} => type_name(&options[0]),
        Schema::Fixed{logical_type_f: None, name, ..} => render_fullname(name),
        Schema::Fixed{logical_type_f: Some(LogicalTypeFixed::DecimalF(d)), ..} => decimal_name(d),
        Schema::Fixed{logical_type_f: Some(LogicalTypeFixed::Duration), ..} => "duration".to_string(),
        Schema::Enum{name, ..} => render_fullname(name),
    }
}

impl Schema {
    pub fn parse_json(av: &JsonValue) -> Result<Schema> {
        parse_schema_json(None, av)
    }
}
pub fn parse_schema_json(context: Option<&TypeName>, av: &JsonValue) -> Result<Schema> { 
    fn parse_fixed(context: Option<&TypeName>, o: &JsonValue) -> Result<Schema> {
        let name = o["name"].as_str().unwrap() ;
        let namespace = o["namespace"].as_str() ;
        let type_name = mk_type_name(context, name, namespace) ;
        let mk_alias = |name: &str| mk_type_name(Some(&type_name), name, None) ;
        let aliases = mk_aliases(&type_name, &serde_json::from_value::<Vec<String>>(o["aliases"].clone()).unwrap_or(vec![])[..]) ;
        let size = o["size"].as_u64().unwrap() ;
        Ok(Schema::Fixed{name: type_name, aliases, size, logical_type_f: None})
    }
    match av {
        JsonValue::String(s) => match s.as_str() {
            "null" => Ok(Schema::Null),
            "boolean" => Ok(Schema::Boolean),
            "int" => Ok(Schema::Int {logical_type_i: None}),
            "long" => Ok(Schema::Long {logical_type_l: None}),
            "float" => Ok(Schema::Float),
            "double" => Ok(Schema::Double),
            "bytes" => Ok(Schema::Bytes {logical_type_b: None}),
            "string" => Ok(Schema::String {logical_type_s: None}),
            "uuid" => Ok(Schema::String {logical_type_s: Some(LogicalTypeString::UUID)}),
            "date" => Ok(Schema::Int {logical_type_i: Some(LogicalTypeInt::Date)}),
            "time-millis" => Ok(Schema::Int {logical_type_i: Some(LogicalTypeInt::TimeMillis)}),
            "time-micros" => Ok(Schema::Long {logical_type_l: Some(LogicalTypeLong::TimeMicros)}),
            "timestamp-millis" => Ok(Schema::Long {logical_type_l: Some(LogicalTypeLong::TimestampMillis)}),
            "timestamp-micros" => Ok(Schema::Long {logical_type_l: Some(LogicalTypeLong::TimestampMicros)}),
            "local-timestamp-millis" => Ok(Schema::Long {logical_type_l: Some(LogicalTypeLong::LocalTimestampMillis)}),
            "local-timestamp-micros" => Ok(Schema::Long {logical_type_l: Some(LogicalTypeLong::LocalTimestampMicros)}),
            somename => Ok(Schema::NamedType(mk_type_name(context, somename, None)))
        },
        JsonValue::Array(arr) => {
            if arr.len() > 0 {
                Ok(Schema::Union{options: arr.into_iter().map(|it| parse_schema_json(context, it)).try_collect()?})
            } else {
                Err(anyhow!("Unions must have at least one type."))
            }
        },
        o@JsonValue::Object(_) => {
            let logical_type = o["logical_type"].as_str() ;
            let ty = o["type"].as_str().unwrap() ;


            match logical_type {
                Some("decimal") => {
                    let precision = o["precision"].as_u64().unwrap() ;
                    let scale = o["scale"].as_u64().unwrap_or(0) ;
                    let dec = Decimal { precision, scale } ;
                    match ty {
                        "bytes" => Ok(Schema::Bytes{logical_type_b: Some(LogicalTypeBytes::DecimalB(dec))}),
                        "fixed" =>  {
                            match parse_fixed(context, o) {
                                Ok(Schema::Fixed {name, aliases, size, ..}) => {
                                    Ok(Schema::Fixed {name, aliases, size, logical_type_f: Some(LogicalTypeFixed::DecimalF(dec))})
                                },
                                x => x,
                            }
                        },
                        "int" => Ok(Schema::Int{logical_type_i: Some(LogicalTypeInt::DecimalI(dec))}),
                        "long" => Ok(Schema::Long{logical_type_l: Some(LogicalTypeLong::DecimalL(dec))}),
                        s => Err(anyhow!(format!("Unsupported underlying type: {:?}", s)))
                    }
                },
                Some("uuid") => {
                    match ty {
                        "string" => Ok(Schema::String {logical_type_s: Some(LogicalTypeString::UUID)}),
                        s => Err(anyhow!(format!("Unsupported underlying type: {}", s)))
                    }
                },
                Some("date") => {
                    match ty {
                        "int" => Ok(Schema::Int {logical_type_i: Some(LogicalTypeInt::Date)}),
                        s => Err(anyhow!(format!("Unsupported underlying type: {}", s)))
                    }
                },
                Some("time-millis") => {
                    match ty {
                        s => Err(anyhow!(format!("Unsupported underlying type: {}", s)))
                    }
                },
                Some("time-micros") => {
                    match ty {
                        s => Err(anyhow!(format!("Unsupported underlying type: {}", s)))
                    }
                },
                Some("timestamp-millis") => {
                    match ty {
                        s => Err(anyhow!(format!("Unsupported underlying type: {}", s)))
                    }
                },
                Some("timestamp-micros") => {
                    match ty {
                        s => Err(anyhow!(format!("Unsupported underlying type: {}", s)))
                    }
                },
                Some("local-timestamp-millis") => {
                    match ty {
                        s => Err(anyhow!(format!("Unsupported underlying type: {}", s)))
                    }
                },
                Some("local-timestamp-micros") => {
                    match ty {
                        s => Err(anyhow!(format!("Unsupported underlying type: {}", s)))
                    }
                },
                Some("duration") => {
                    match ty {
                        s => Err(anyhow!(format!("Unsupported underlying type: {}", s)))
                    }
                },
                Some(_) => Ok(serde_json::from_str::<Schema>(&ty).unwrap()),
                None => {
                    match ty {
                        "map" => {
                            Ok(Schema::Map{values: Box::new(parse_schema_json(context, &o["values"].clone())?)})
                        },
                        "array" => {
                            Ok(Schema::Array {item: Box::new(parse_schema_json(context, &o["items"].clone())?)})
                        },
                        "record" => {
                            let name = o["name"].as_str().unwrap() ;
                            let namespace = o["namespace"].as_str() ;
                            let type_name = mk_type_name(context, name, namespace) ;
                            trace!("parse_schema_json => record name: {:?}", type_name) ;
                            let aliases = mk_aliases(&type_name, &serde_json::from_value::<Vec<String>>(o["aliases"].clone()).unwrap_or(vec![])[..]) ;
                            let doc = o["doc"].as_str().map(|x| x.to_string()) ;
                            let fields = o["fields"].as_array().unwrap().into_iter().map(|it| {
                                parse_field(&type_name, it)
                            }).try_collect()? ;
                            Ok(Schema::Record{name: type_name, aliases, doc, fields})
                        },
                        "enum" => {
                            let name = o["name"].as_str().unwrap() ;
                            let namespace = o["namespace"].as_str() ;
                            let type_name = mk_type_name(context, name, namespace) ;
                            trace!("parse_schema_json => enum name: {:?}", type_name) ;
                            let aliases = mk_aliases(&type_name, &serde_json::from_value::<Vec<String>>(o["aliases"].clone()).unwrap_or(vec![])[..]) ;
                            let doc = o["doc"].as_str().map(|x| x.to_string()) ;
                            let symbols = serde_json::from_value::<Vec<String>>(o["symbols"].clone()).unwrap() ;
                            Ok(Schema::Enum { name: type_name, aliases, doc, symbols })
                        },
                        "fixed" => parse_fixed(context, o), 
                        "null" => Ok(Schema::Null),
                        "boolean" => Ok(Schema::Boolean),
                        "int" => Ok(Schema::Int{logical_type_i: None}),
                        "long" => Ok(Schema::Long{logical_type_l: None}),
                        "float" => Ok(Schema::Float),
                        "double" => Ok(Schema::Double),
                        "bytes" => Ok(Schema::Bytes{logical_type_b: None}),
                        "string" => Ok(Schema::String{logical_type_s: None}),
                        s => Err(anyhow!(format!("typeMismatch: Unrecognized object type: {}", s)))
                    }
                },
            }
        },
        invalid => Err(anyhow!(format!("Invalid JSON for Avro Schema: {:?}", invalid)))
    }
}

pub fn mk_aliases(context: &TypeName, names: &[String]) -> Vec<TypeName> {
    names.into_iter().map(|name| mk_type_name(Some(context), name, None) ).collect_vec()
}

pub fn parse_field(record: &TypeName, av: &JsonValue) -> Result<Field> {
    match av {
        o@JsonValue::Object(_) => {
            let fld_name = o["name"].as_str().unwrap().to_string() ;
            let fld_doc = o["doc"].as_str().map(|x| x.to_string()) ;
            let fld_type = parse_schema_json(Some(record), &o["type"])? ;
            let err = |schema: &Schema, e: &TypeName| -> Result<Schema> {
                Err(anyhow!(format!("Haskell Avro bindings does not support default for aliased or recursive types at this time.")))
            } ;
            let def_m = o.get("default");
            let fld_default = match def_m {
                Some(x) => {
                    match parse_field_default(&err, &fld_type, x) {
                        Ok(r) => Some(r),
                        Err(e) => return Err(anyhow!(format!("{}", e))) ,
                    }
                },
                None => None,
            };
            let fld_order = serde_json::from_value::<Order>(o["order"].clone()).ok().or(Some(Order::Ascending)) ;

            let fld_aliases = serde_json::from_value::<Vec<String>>(o["aliases"].to_owned()).unwrap_or(vec![]) ;
            Ok(Field{fld_name, fld_aliases, fld_doc, fld_order, fld_type, fld_default})
        },
        invalid => Err(anyhow!(format!("typeMismatch: Field: {:?}", invalid)))
    }
}

pub fn parse_field_default<F>(env: &F, schema: &Schema, value: &JsonValue) -> Result<DefaultValue> 
where
    F: Fn(&Schema, &TypeName) -> Result<Schema> 
{
    let default_union = |schema: &Schema, value: &JsonValue| -> Result<DefaultValue> {
        match (schema, value) {
            (Schema::Union{ options, ..}, val) => Ok(DefaultValue::DUnion (options.clone(), options[0].clone(), Box::new(parse_field_default(env, &options[0], val)?))),
            (_, _) => Err(anyhow!(format!("Impossible: not Union.")))
        }
    } ;
    parse_avro_json(&default_union, env, schema, value)
}

pub fn parse_avro_json<U, V>(union: &U, env: &V, ty: &Schema, av: &JsonValue) -> Result<DefaultValue>
where
    U : Fn(&Schema, &JsonValue) -> Result<DefaultValue>,
    V : Fn(&Schema, &TypeName) -> Result<Schema>
{
    match (&av, &ty) {
        (_, Schema::NamedType(name)) => {
            match env(&ty, &name) {
                Err(_) => Err(anyhow!(format!("Could not resolve type name for {}", render_fullname(&name)))),
                Ok(t) => parse_avro_json(union, env, &t, av)
            }
        },
        (_, u@Schema::Union{..}) => {
            union(u, av)
        },

        (JsonValue::String(s), Schema::String {..}) => Ok(DefaultValue::DString(ty.to_owned(), s.to_owned())),
        (JsonValue::String(s), Schema::Enum {name, symbols, ..}) => match symbols.iter().position(|x| x == s) {
            Some(i) => Ok(DefaultValue::DEnum(ty.clone(), i, s.to_owned())),
            None => Err(anyhow!(format!("JSON string is not one of the expected symbols for enum {:?}: {:?})", name, s)))
        },
        (JsonValue::String(s), Schema::Bytes { .. }) => Ok(DefaultValue::DBytes(ty.to_owned(), s.to_owned().into_bytes())),
        (JsonValue::String(s), Schema::Fixed { size, .. }) => {
            let bytes = s.to_owned().into_bytes() ;
            let len = bytes.len() as u64;
            if &len != size {
                return Err(anyhow!(format!("Fixed string wrong size. Expected {:?} but got {:?}", size, len)))
            }
            Ok(DefaultValue::DFixed(ty.to_owned(), bytes))
        },
        (JsonValue::String(_), _) => Err(anyhow!(format!("Expected type String, Enum, Bytes, or Fixed, but found (Type,Value)=({:?}, {:?})", ty, av))),

        (JsonValue::Bool(b), Schema::Boolean) => Ok(DefaultValue::DBoolean(b.to_owned())),
        (JsonValue::Bool(_), _) => avro_type_mismatch(&ty, "boolean"),

        (JsonValue::Number(i), Schema::Int { .. }) => Ok(DefaultValue::DInt(ty.to_owned(), i.as_i64().unwrap() as i32)),
        (JsonValue::Number(i), Schema::Long { .. }) => Ok(DefaultValue::DLong(ty.to_owned(), i.as_i64().unwrap())),
        (JsonValue::Number(i), Schema::Float { .. }) => Ok(DefaultValue::DFloat(ty.to_owned(), i.as_f64().unwrap() as f32)),
        (JsonValue::Number(i), Schema::Double { .. }) => Ok(DefaultValue::DDouble(ty.to_owned(), i.as_f64().unwrap())),
        (JsonValue::Number(_), _) => avro_type_mismatch(&ty, "number"),

        (JsonValue::Array(vec), Schema::Array { item, .. }) => {
            let val = vec.into_iter().map(|x| parse_avro_json(union, env, item, x)).try_collect() ;
            val.map(DefaultValue::DArray)
        },

        (JsonValue::Array(_), _) => avro_type_mismatch(&ty, "array"),

        (JsonValue::Object(obj), Schema::Map { values, .. }) => {
            let val = obj.into_iter().map(|(k, v)| {
                parse_avro_json(union, env, values, v).map(|it| (k.to_owned(), it))
            }).try_collect() ;
            val.map(DefaultValue::DMap)
        },

        (JsonValue::Object(obj), Schema::Record { fields, .. }) => {
            let lk_and_parse = |f: &Field| {
                match obj.get(&f.fld_name) {
                    Some(v) => {
                        debug!("parse_avro_json => field: {:?}", &f.fld_name) ;
                        parse_avro_json(union, env, &f.fld_type, v)
                    },
                    None => match &f.fld_default {
                        Some(v) => Ok(v.to_owned()),
                        None => Err(anyhow!(format!("Decode failure: No record field '{}' and no default in schema.", &f.fld_name)))
                    }
                }
            } ;
            let val = fields.into_iter().map(|f| lk_and_parse(&f).map(|it| (f.fld_name.to_owned(), it))).try_collect()? ;
            Ok(DefaultValue::DRecord(ty.to_owned(), val))
        },

        (JsonValue::Object(_), _) => avro_type_mismatch(&ty, "null"),


        (JsonValue::Null, Schema::Null) => Ok(DefaultValue::DNull),
        (JsonValue::Null, _) => avro_type_mismatch(&ty, "null"),
    }
}

pub fn avro_type_mismatch<T>(expected: &Schema, actual: &str) -> Result<T> {
    Err(anyhow!(format!("Could not resolve type '{:?}' with expected type:  {:?}", expected, actual)))
}

pub fn build_type_environment<F>(failure: F, schema: &Schema, name: &TypeName) -> Result<Schema>
where
    F: Fn(&TypeName) -> Result<Schema>
{
    let env = extract_bindings(schema) ;
    match env.get(name) {
        None => failure(name),
        Some(res) => Ok(res.to_owned())
    }
}

pub fn extract_bindings(schema: &Schema) -> HashMap<TypeName, Schema> {
    match schema {
        t@Schema::Record { name, aliases, fields, ..} => {
            let with_record = iter::once(name).chain(aliases.into_iter()).zip(iter::repeat(t)).map(|(k, v)| (k.to_owned(), v.to_owned())) ;
            with_record.chain(fields.into_iter().flat_map(|it| extract_bindings(&it.fld_type))).collect()
        },
        e@Schema::Enum { name, aliases, ..} => {
            iter::once(name).chain(aliases.into_iter()).zip(iter::repeat(e)).map(|(k, v)| (k.to_owned(), v.to_owned())).collect()
        },
        Schema::Union { options, ..} => {
            options.into_iter().flat_map(|it| extract_bindings(it)).collect()
        },

        f@Schema::Fixed { name, aliases, ..} => {
            iter::once(name).chain(aliases.into_iter()).zip(iter::repeat(f)).map(|(k, v)| (k.to_owned(), v.to_owned())).collect()
        },
        Schema::Array { item, ..} => extract_bindings(item),
        Schema::Map { values, ..} => extract_bindings(values),
        Schema::Null | Schema::Boolean |
        Schema::Int{..} | Schema::Long{..} | 
        Schema::Float | Schema::Double | 
        Schema::Bytes {..} | Schema::String {..} |
        Schema::NamedType{..} 
        => HashMap::new()
    }
}

